{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# asyncio ：python异步IO\n",
    "## 协程与异步\n",
    "### 协程\n",
    "Coroutines使用 saync/await 写法，例如：  \n",
    "\n",
    "注意jupyter中不能使用asyncio.run(func()),使用await func()代替，并用##注释  \n",
    "note:jupyter中需要运行过的代码块中的内容才会存在。跳过运行会出现import error 或者变量函数未定义错误\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import time\n",
    "\n",
    "async def say_after(delay, what):\n",
    "    await asyncio.sleep(delay)\n",
    "    print(what)\n",
    "    print(f\"print {what} at {time.strftime('%X')}\")\n",
    "    return what"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "started at 22:31:33\nhello\nprint hello at 22:31:34\nworld\nprint world at 22:31:36\nfinished at 22:31:36\n"
    }
   ],
   "source": [
    "async def f1():\n",
    "    print(f\"started at {time.strftime('%X')}\")\n",
    "\n",
    "    await say_after(1, 'hello')\n",
    "    await say_after(2, 'world')\n",
    "\n",
    "    print(f\"finished at {time.strftime('%X')}\")\n",
    "\n",
    "await f1() ##"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "要真正运行一个协程，asyncio 提供了三种主要机制:\n",
    "+ `asyncio.run()` 函数用来运行最高层级的入口点 \"`main()`\" 函数 (参见上面的示例。)；\n",
    "+ 等待一个协程。以上代码段会在等待 1 秒后打印 \"hello\"，然后 再次 等待 2 秒后打印 \"world\"；\n",
    "+ `asyncio.create_task()` 函数用来并发运行作为 `asyncio` 任务 的多个协程。："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "started at 17:08:49\nhello\nprint hello at 17:08:50\nasyncio\nprint asyncio at 17:08:51\nfinished at 17:08:51\n"
    }
   ],
   "source": [
    "async def f2():\n",
    "    '''task'''\n",
    "    task1 = asyncio.create_task(say_after(1, 'hello'))\n",
    "    task2 = asyncio.create_task(say_after(2, 'asyncio'))\n",
    "    print(f\"started at {time.strftime('%X')}\")\n",
    "    await task1\n",
    "    await task2\n",
    "    print(f\"finished at {time.strftime('%X')}\")\n",
    "\n",
    "await f2() ##\n",
    "#注意 task2 在函数开始后等待两秒而非 task1 结束后等待\n",
    "#即task2与task1同时开始"
   ]
  },
  {
   "cell_type": "markdown",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### 可等待对象 \n",
    "\n",
    "如果一个对象可以在`await`语句中使用，那么它就是 可等待 对象。许多 asyncio API 都被设计为接受可等待对象。\n",
    "\n",
    " 1. 协程：  \n",
    "    协程属于可等待对象，故可在其他协程中被等待 \n",
    ">在本文档中 \"协程\" 可用来表示两个紧密关联的概念:\n",
    ">+ 协程函数: 定义形式为 `async def` 的函数;\n",
    ">+ 协程对象: 调用 协程函数 所返回的对象。\n",
    "\n",
    " 2. 任务  \n",
    "    任务 被用来设置日程以便 并发 执行协程。  \n",
    "    当一个协程通过 asyncio.create_task() 等函数被打包为一个 任务，该协程将自动排入日程准备立即运行:\n",
    " 3. `future`对象  \n",
    "`Future` 是一种特殊的 低层级 可等待对象，表示一个异步操作的 最终结果。  \n",
    "当一个 `Future` 对象 被等待，这意味着协程将保持等待直到该 `Future` 对象在其他地方操作完毕。  \n",
    "在 `asyncio` 中需要 `Future` 对象以便允许通过 `async`/`await` 使用基于回调的代码。  \n",
    "通常情况下 没有必要 在应用层级的代码中创建 `Future` 对象。  \n",
    "`Future` 对象有时会由库和某些 `asyncio` API 暴露给用户，用作可等待对象\n",
    "\n",
    "\n",
    " ### 运行 asyncio 程序\n",
    "\n",
    "`asyncio.run(coro, *, debug=False)`\n",
    "\n",
    "​\t执行 coroutine coro 并返回结果。\n",
    "​\t此函数运行传入的协程，负责管理 asyncio 事件循环并 完结异步生成器。\n",
    "​\t当有其他 asyncio 事件循环在同一线程中运行时，此函数不能被调用。\n",
    "​\t如果 debug 为 True，事件循环将以调试模式运行。\n",
    "​\t此函数总是会创建一个新的事件循环并在结束时关闭之。它应当被用作 asyncio 程序的主入口点，理想情况下应当只被调用一次。\n",
    "\n",
    "### 创建任务   \n",
    "\n",
    "`asyncio.create_task(coro, *, name=None)`  \n",
    "\t将 coro 协程 打包为一个 Task 排入日程准备执行。返回 Task 对象。\n",
    "\t如果name不空，则传入`Task.set_name()`将其作为任务名.\n",
    "    该任务会在 `get_running_loop()` 返回的循环中执行，如果当前线程没有在运行的循环则会引发 RuntimeError。\n",
    "    此函数 在 Python 3.7 中被加入。在 Python 3.7 之前，可以改用低层级的 `asyncio.ensure_future()` 函数。\n",
    "\n",
    "### 休眠\n",
    "\n",
    "`coroutine* asyncio.sleep(delay, result=None, *, loop=None`)`\n",
    "\n",
    "  阻塞*delay*指定的秒数。 如果指定了*result*，则当协程完成时将其返回给调用者。 `sleep()` 总是会挂起当前任务，以允许其他任务运行。  \n",
    "Deprecated since version 3.8, will be removed in version 3.10: *loop* 形参。 \n",
    "以下协程示例运行 5 秒，每秒显示一次当前日期:\n",
    "\n",
    "\n",
    "\n",
    "### 并发运行任务\n",
    "\n",
    "`awaitable asyncio.gather(*aws, loop=None, return_exceptions=False)`\n",
    "\n",
    "  *并发* 运行 *aws* 序列中的 [可等待对象](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio-awaitables)。  \n",
    "  如果 *aws* 中的某个可等待对象为协程，它将自动作为一个任务加入日程。   \n",
    "  如果所有可等待对象都成功完成，结果将是一个由所有返回值聚合而成的列表。结果值的顺序与 *aws* 中可等待对象的顺序一致。  \n",
    "如果 *return_exceptions* 为 `False` (默认)，所引发的首个异常会立即传播给等待 `gather()` 的任务。*aws* 序列中的其他可等待对象 **不会被取消** 并将继续运行。 如果 *return_exceptions* 为 `True`，异常会和成功的结果一样处理，并聚合至结果列表。  \n",
    " 如果 `gather()` *被取消*，所有被提交 (尚未完成) 的可等待对象也会 *被取消*。 如果 *aws* 序列中的任一`Task`或`Future`对象 *被取消*，它将被当作引发了 [`CancelledError`](https://docs.python.org/zh-cn/3/library/asyncio-exceptions.html#asyncio.CancelledError) 一样处理 -- 在此情况下 `gather()` 调用 **不会** 被取消。这是为了防止一个已提交的 Task`/`Future 被取消导致其他 `Tasks`/`Future` 也被取消。  \n",
    "Deprecated since version 3.8, will be removed in version 3.10: *loop* 形参。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "hello\nprint hello at 22:40:43\nasyncio\nprint asyncio at 22:40:44\n"
    }
   ],
   "source": [
    "async def f3():\n",
    "    await asyncio.gather(say_after(1, 'hello'), say_after(2, 'asyncio'))\n",
    "\n",
    "await f3()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 屏蔽取消操作\n",
    "\n",
    "`awaitable asyncio.shield(aw, *, loop=None)`\n",
    "\n",
    "  保护一个 [可等待对象](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio-awaitables) 防止其被 [`取消`](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio.Task.cancel)。 如果 *aw* 是一个协程，它将自动作为任务加入日程。  \n",
    "以下语句:  \n",
    " `res = await shield(something()) `  \n",
    "  相当于:   \n",
    "  `res = await something() `  \n",
    "   *不同之处* 在于如果包含它的协程被取消，在 `something()` 中运行的任务不会被取消。  \n",
    "   从 `something()` 的角度看来，取消操作并没有发生。然而其调用者已被取消，因此 \"await\" 表达式仍然会引发 [`CancelledError`](https://docs.python.org/zh-cn/3/library/asyncio-exceptions.html#asyncio.CancelledError)。 如果通过其他方式取消 `something()` (例如在其内部操作) 则 `shield()` 也会取消。   \n",
    "如果希望完全忽略取消操作 (不推荐) 则 `shield()` 函数需要配合一个 try/except 代码段，如下所示:  \n",
    " `try:    res = await shield(something()) except CancelledError:    res = None `   \n",
    "   Deprecated since version 3.8, will be removed in version 3.10: *loop* 形参。\n",
    "\n",
    "   ## 超时\n",
    "\n",
    "`coroutine asyncio.wait_for(aw, timeout, *, loop=None)`\n",
    "\n",
    "  等待 *aw* [可等待对象](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio-awaitables) 完成，指定 timeout 秒数后超时。  \n",
    "  如果 *aw* 是一个协程，它将自动作为任务加入日程。  \n",
    "  *timeout* 可以为 `None`，也可以为 float 或 int 型数值表示的等待秒数。如果 *timeout* 为 `None`，则等待直到完成。 如果发生超时，任务将取消并引发 [`asyncio.TimeoutError`](https://docs.python.org/zh-cn/3/library/asyncio-exceptions.html#asyncio.TimeoutError).  \n",
    "   要避免任务 [`取消`](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio.Task.cancel)，可以加上 [`shield()`](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio.shield)。   \n",
    "   函数将等待直到目标对象确实被取消，所以总等待时间可能超过 *timeout* 指定的秒数。  \n",
    "   如果等待被取消，则 *aw* 指定的对象也会被取消。  \n",
    " Deprecated since version 3.8, will be removed in version 3.10: *loop* 形参。  \n",
    " 在 3.7 版更改: 当 aw 因超时被取消，`wait_for` 会等待 *aw* 被取消。之前版本则将立即引发 `asyncio.TimeoutError`。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "f5 timeout\nf54 timeout\nsleep for 4 second\n"
    }
   ],
   "source": [
    "async def sleep_4():\n",
    "    await asyncio.sleep(3)\n",
    "    print('sleep for 4 second')\n",
    "\n",
    "async def f4():\n",
    "    try:\n",
    "        await asyncio.wait_for(sleep_4(), timeout=2)\n",
    "    except asyncio.TimeoutError:\n",
    "        print('f4 timeout')\n",
    "        await asyncio.sleep(2)\n",
    "\n",
    "\n",
    "async def f4_shield():\n",
    "    '''加入sheild后sleep_4在超时后依旧运行'''\n",
    "    try:\n",
    "        res = asyncio.shield(sleep_4())\n",
    "        await asyncio.wait_for(res, timeout=2)\n",
    "    except asyncio.TimeoutError:\n",
    "        print('f4_shield timeout')\n",
    "        await asyncio.sleep(2)\n",
    "\n",
    "await f4()\n",
    "await f4_shield()"
   ]
  },
  {
   "cell_type": "markdown",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### 简单等待\n",
    "\n",
    "`coroutine asyncio.wait(aws, *, loop=None, timeout=None, return_when=ALL_COMPLETED)`\n",
    "\n",
    "  并发运行 *aws* 指定的 [可等待对象](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio-awaitables) 并阻塞线程直到满足 `return_when` 指定的条件。 \n",
    "   返回两个 `Task`/`Future` 集合: `(done, pending)`。 用法:   \n",
    "   `done, pending = await asyncio.wait(aws) `  \n",
    "    如指定 *timeout* (float 或 int 类型) 则它将被用于控制返回之前等待的最长秒数。  \n",
    "     请注意此函数不会引发 [`asyncio.TimeoutError`](https://docs.python.org/zh-cn/3/library/asyncio-exceptions.html#asyncio.TimeoutError)。当超时发生时，未完成的 Future 或 Task 将在指定秒数后被返回。    \n",
    "     *return_when* 指定此函数应在何时返回。它必须为以下常数之一:  \n",
    "     \n",
    "| 常数              | 描述                                                         |\n",
    "| ----------------- | ------------------------------------------------------------ |\n",
    "| `FIRST_COMPLETED` | 函数将在任意可等待对象结束或取消时返回。                     |\n",
    "| `FIRST_EXCEPTION` | 函数将在任意可等待对象因引发异常而结束时返回。当没有引发任何异常时它就相当于 `ALL_COMPLETED`。 |\n",
    "| `ALL_COMPLETED`   | 函数将在所有可等待对象结束或取消时返回。                     |    \n",
    "               \n",
    "  与 [`wait_for()`](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio.wait_for) 不同，`wait()` 在超时发生时不会取消可等待对象。  \n",
    "  3.8 版后已移除: 如果 *aws* 中的某个可等待对象为协程，它将自动作为任务加入日程。直接向 `wait()` 传入协程对象已弃用，因为这会导致 [令人迷惑的行为](https://docs.python.org/zh-cn/3/library/asyncio-task.html#asyncio-example-wait-coroutine)。  \n",
    "    Deprecated since version 3.8, will be removed in version 3.10: *loop* 形参。   \n",
    "     注解 `wait()` 会自动将协程作为任务加入日程，以后将以 `(done, pending)` 集合形式返回显式创建的任务对象。  \n",
    "     3.8 版后已移除: 直接向 wait() 传入协程对象的方式已弃用。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "1\nprint 1 at 10:19:47\ntask in done\n"
    }
   ],
   "source": [
    "async def f6():\n",
    "    task = asyncio.create_task(say_after(1, '1'))\n",
    "    done, pending = await asyncio.wait({task}, timeout=2.1)\n",
    "    if task in done:\n",
    "        print('task in done')\n",
    "\n",
    "await f6()    "
   ]
  },
  {
   "cell_type": "markdown",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "     `asyncio.as_completed(aws, *, loop=None, timeout=None)`\n",
    "\n",
    "并发地运行 *aws* 集合中的 *可等待对象*。返回一个 `Future` 对象的迭代器。返回的每个 `Future` 对象代表来自剩余可等待对象集合的最早结果。  \n",
    "  如果在所有 `Future` 对象完成前发生超时则将引发 `asyncio.TimeoutError`。  \n",
    "  Deprecated since version 3.8, will be removed in version 3.10: loop 形参。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": "1\nprint 1 at 15:10:27\n1\n"
    }
   ],
   "source": [
    "async def f7():\n",
    "    task = asyncio.create_task(say_after(1, '1'))\n",
    "    for i in asyncio.as_completed({task}):\n",
    "        a = await i\n",
    "        print(a)\n",
    "await f7()\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### 来自其他线程的日程安排\n",
    "`asyncio.run_coroutine_threadsafe(coro, loop)`  \n",
    "\n",
    " 向指定事件循环提交一个协程。线程安全。  \n",
    " 返回一个 `concurrent.futures.Future` 以等待来自其他 OS 线程的结果。  \n",
    " 此函数应该从另一个 OS 线程中调用，而非事件循环运行所在线程。示例:\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a coroutine\n",
    "coro = asyncio.sleep(1, result=3)\n",
    "loop=asyncio.new_event_loop()\n",
    "# Submit the coroutine to a given loop\n",
    "future = asyncio.run_coroutine_threadsafe(coro, loop)\n",
    "\n",
    "# Wait for the result with an optional timeout argument\n",
    "assert future.result() == 3\n",
    "#如果在协程内产生了异常，将会通知返回的 Future 对象。它也可被用来取消事件循环中的任务:\n",
    "try:\n",
    "    result = future.result(4)\n",
    "    print(result)\n",
    "except asyncio.TimeoutError:\n",
    "    print('The coroutine took too long, cancelling the task...')\n",
    "    future.cancel()\n",
    "except Exception as exc:\n",
    "    print(f'The coroutine raised an exception: {exc!r}')\n",
    "else:\n",
    "    print(f'The coroutine returned: {result!r}')\n",
    "#不同与其他 asyncio 函数，此函数要求显式地传入 loop 参数。"
   ]
  },
  {
   "cell_type": "markdown",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### 内省\n",
    "\n",
    "`asyncio.current_task(loop=None)`  \n",
    "  返回当前运行的 Task 实例，如果没有正在运行的任务则返回 None。  \n",
    "如果 loop 为 None 则会使用 get_running_loop() 获取当前事件循环。  \n",
    "    3.7 新版功能.  \n",
    "    \n",
    "`asyncio.all_tasks(loop=None)`  \n",
    " 返回事件循环所运行的未完成的 `Task` 对象的集合。  \n",
    "    如果 `loop` 为 `None`，则会使用 `get_running_loop()` 获取当前事件循环。  \n",
    "    3.7 新版功能.\n",
    "\n",
    "### Task 对象\n",
    "\n",
    "`class asyncio.Task(coro, *, loop=None, name=None)`  \n",
    "    一个与 `Future` 类似 的对象，可运行 *Python* 协程。*非线程安全*。 \n",
    "  `Task` 对象被用来在事件循环中运行协程。如果一个协程在等待一个 `Future` 对象，`Task` 对象会挂起该协程的执行并等待该 `Future` 对象完成。当该 `Future` 对象 完成，被打包的协程将恢复执行。  \n",
    "  事件循环使用协同日程调度: 一个事件循环每次运行一个 `Task` 对象。而一个 `Task` 对象会等待一个 `Future` 对象完成，该事件循环会运行其他 `Task`、回调或执行 IO 操作。  \n",
    "  使用高层级的 `asyncio.create_task()` 函数来创建 `Task` 对象，也可用低层级的 `loop.create_task()` 或 `ensure_future()` 函数。不建议手动实例化 `Task` 对象。  \n",
    "  要取消一个正在运行的 `Task` 对象可使用 `cancel()` 方法。调用此方法将使该 `Task` 对象抛出一个 `CancelledError` 异常给打包的协程。如果取消期间一个协程正在等待一个 `Future` 对象，该 `Future` 对象也将被取消。  \n",
    "  `cancelled()` 可被用来检测 `Task` 对象是否被取消。如果打包的协程没有抑制 `CancelledError` 异常并且确实被取消，该方法将返回 `True`。  \n",
    "  `asyncio.Task` 从 `Future` 继承了其除 `Future.set_result()` 和 `Future.set_exception()` 以外的所有 API。\n",
    "  `Task` 对象支持 `contextvars` 模块。当一个 `Task` 对象被创建，它将复制当前上下文，然后在复制的上下文中运行其协程。  \n",
    "  在 3.7 版更改: 加入对 `contextvars` 模块的支持。   \n",
    "    在 3.8 版更改: Added the name parameter.  \n",
    "    Deprecated since version 3.8, will be removed in version 3.10: loop 形参。\n",
    "\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "file_extension": ".py",
  "kernelspec": {
   "display_name": "Python 3.8",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.1-final"
  },
  "mimetype": "text/x-python",
  "name": "python",
  "npconvert_exporter": "python",
  "pygments_lexer": "ipython3",
  "version": 3
 },
 "nbformat": 4,
 "nbformat_minor": 4
}